package cn.lokei.gateway.handler.mqtt;

import java.io.IOException;
import java.util.List;

import javax.annotation.PostConstruct;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.client.RestTemplate;

import cn.lokei.gateway.entity.IotChannel;
import cn.lokei.gateway.handler.message.ReceiveHandler;
import cn.lokei.gateway.service.ChannelSupervise;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.codec.mqtt.MqttConnectPayload;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttSubscribePayload;
import io.netty.handler.codec.mqtt.MqttTopicSubscription;
import io.netty.handler.codec.mqtt.MqttUnsubscribeMessage;

public class MqttHandler extends ChannelInboundHandlerAdapter {

  private Logger log = LoggerFactory.getLogger(this.getClass());

  private ChannelSupervise channelSupervise = ChannelSupervise.getBean(ChannelSupervise.class);

  private RestTemplate restTemplate = new RestTemplate();

  private ReceiveHandler receiveHandler = new ReceiveHandler();

  // @Autowired
  // KafkaProducer kafkaProducer;

  /**
   * 客户端与服务端第一次建立连接时执行 在channelActive方法之前执行
   */
  @Override
  public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
    super.channelRegistered(ctx);
    channelSupervise.addChannel(ctx.channel());
  }

  /**
   * 客户端与服务端 断连时执行 channelInactive方法之后执行
   */
  @Override
  public void channelUnregistered(ChannelHandlerContext ctx) throws Exception {
    super.channelUnregistered(ctx);
    channelSupervise.removeChannel(ctx.channel());
  }

  /**
   * 从客户端收到新的数据时，这个方法会在收到消息时被调用
   */
  @Override
  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception, IOException {
    MqttMessage mqttMessage = (MqttMessage) msg;
    log.info("info--" + mqttMessage.toString());
    MqttFixedHeader mqttFixedHeader = mqttMessage.fixedHeader();
    Channel channel = ctx.channel();

    if (mqttFixedHeader.messageType().equals(MqttMessageType.CONNECT)) {
      // 在一个网络连接上，客户端只能发送一次CONNECT报文。服务端必须将客户端发送的第二个CONNECT报文当作协议违规处理并断开客户端的连接
      // to do 建议connect消息单独处理，用来对客户端进行认证管理等 这里直接返回一个CONNACK消息
      MqttConnectPayload payload = (MqttConnectPayload) mqttMessage.payload();
      IotChannel iotChannel = channelSupervise.findChannel(channel.id().asLongText());
      if(iotChannel != null) {
        iotChannel.set_connected(true);
        iotChannel.setClient_id(payload.clientIdentifier());
        iotChannel.setUsername(payload.userName());
        iotChannel.setPassword(payload.passwordInBytes());
        try {
          // RestTemplate restTemplate = RestTemplate.getBean("restTemplate");
          // kafkaProducer.send(mqttMessage);
          ResponseEntity<String> responseEntity = restTemplate.getForEntity("https://iot.lokei.cn/auth/device",
            String.class);
          HttpStatus statusCode = responseEntity.getStatusCode(); // 获取响应码
          if (statusCode == HttpStatus.OK) {
            iotChannel.set_registered(true);
          } else {
            iotChannel.set_registered(true);
          }
        } catch (Exception e) {
          iotChannel.set_registered(false);
        }
      }
      MqttMsgBack.connack(channel, mqttMessage);
    }

    switch (mqttFixedHeader.messageType()) {
      case PUBLISH: // 客户端发布消息
        // PUBACK报文是对QoS 1等级的PUBLISH报文的响应
        // System.out.println("123");
        // KafkaProducer kafkaProducer = KafkaProducer.getBean("kafkaProducer");
        // kafkaProducer.send(mqttMessage);
        receiveHandler.receiveMessage(mqttMessage);
        MqttMsgBack.puback(channel, mqttMessage);
        break;
      case PUBREL: // 发布释放
        // PUBREL报文是对PUBREC报文的响应
        // to do
        MqttMsgBack.pubcomp(channel, mqttMessage);
        break;
      case SUBSCRIBE: // 客户端订阅主题
        // 客户端向服务端发送SUBSCRIBE报文用于创建一个或多个订阅，每个订阅注册客户端关心的一个或多个主题。
        // 为了将应用消息转发给与那些订阅匹配的主题，服务端发送PUBLISH报文给客户端。
        // SUBSCRIBE报文也（为每个订阅）指定了最大的QoS等级，服务端根据这个发送应用消息给客户端
        // to do
        MqttSubscribePayload payload = (MqttSubscribePayload) mqttMessage.payload();
        List<MqttTopicSubscription> topicSubscriptions = payload.topicSubscriptions();
        for (MqttTopicSubscription mqttTopicSubscription : topicSubscriptions) {
          channelSupervise.subTopic(mqttTopicSubscription.topicName(), channel);
        }
        MqttMsgBack.suback(channel, mqttMessage);
        break;
      case UNSUBSCRIBE: // 客户端取消订阅
        // 客户端发送UNSUBSCRIBE报文给服务端，用于取消订阅主题
        // to do
        MqttUnsubscribeMessage mqttUnsubscribeMessage = (MqttUnsubscribeMessage) mqttMessage;
        channelSupervise.unsubTopic(mqttUnsubscribeMessage.payload().topics(), channel);
        MqttMsgBack.unsuback(channel, mqttMessage);
        break;
      case PINGREQ: // 客户端发起心跳
        // 客户端发送PINGREQ报文给服务端的
        // 在没有任何其它控制报文从客户端发给服务的时，告知服务端客户端还活着
        // 请求服务端发送 响应确认它还活着，使用网络以确认网络连接没有断开
        MqttMsgBack.pingresp(channel, mqttMessage);
        break;
      case DISCONNECT: // 客户端主动断开连接
        // DISCONNECT报文是客户端发给服务端的最后一个控制报文， 服务端必须验证所有的保留位都被设置为0
        // to do
        break;
      default:
        break;
    }
  }

  /**
   * 从客户端收到新的数据、读取完成时调用
   */
  @Override
  public void channelReadComplete(ChannelHandlerContext ctx) throws IOException {
  }

  /**
   * 当出现 Throwable 对象才会被调用，即当 Netty 由于 IO 错误或者处理器在处理事件时抛出的异常时
   */
  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    super.exceptionCaught(ctx, cause);
    ctx.close();
  }

  /**
   * 客户端与服务端第一次建立连接时执行
   */
  @Override
  public void channelActive(ChannelHandlerContext ctx) throws Exception {
    super.channelActive(ctx);
  }

  /**
   * 客户端与服务端 断连时执行
   */
  @Override
  public void channelInactive(ChannelHandlerContext ctx) throws Exception, IOException {
    super.channelInactive(ctx);
  }

  /**
   * 服务端 当读超时时 会调用这个方法
   */
  @Override
  public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception, IOException {
    super.userEventTriggered(ctx, evt);
    ctx.close();
  }

  @Override
  public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception {
    super.channelWritabilityChanged(ctx);
  }
}
